package com.huawei.hbase.data;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;

public class PutData {
    public static void main(String[] args) throws IOException {
        //链接hbase
        Configuration conf = HBaseConfiguration.create();
        //这里的zookeeper地址要改为自己集群的zookeeper地址
        conf.set("hbase.zookeeper.quorum",
                "192.168.0.151");
        conf.set("hbase.zookeeper.property.clientPort", "2181");
        Connection connection = ConnectionFactory.createConnection(conf);
        Table hTable = connection.getTable(TableName.valueOf("users"));

        //插入一条
        Put put = new Put(Bytes.toBytes("row1"));
        put.addColumn(Bytes.toBytes("f"), Bytes.toBytes("id"), Bytes.toBytes("1"));
        put.addColumn(Bytes.toBytes("f"), Bytes.toBytes("name"), Bytes.toBytes("zjhw"));
        put.addColumn(Bytes.toBytes("f"), Bytes.toBytes("age"), Bytes.toBytes(27));
        put.addColumn(Bytes.toBytes("f"), Bytes.toBytes("phone"), Bytes.toBytes("18238485602"));
        put.addColumn(Bytes.toBytes("f"), Bytes.toBytes("emil"), Bytes.toBytes("123654@qq.com"));
        hTable.put(put);
        //插入多个
        Put put1 = new Put(Bytes.toBytes("row2"));
        put1.addColumn(Bytes.toBytes("f"), Bytes.toBytes("id"), Bytes.toBytes("2"));
        put1.addColumn(Bytes.toBytes("f"), Bytes.toBytes("name"), Bytes.toBytes("zhehua"));

        Put put2 = new Put(Bytes.toBytes("row3"));
        put2.addColumn(Bytes.toBytes("f"), Bytes.toBytes("id"), Bytes.toBytes("3"));
        put2.addColumn(Bytes.toBytes("f"), Bytes.toBytes("name"), Bytes.toBytes("zhejiang"));

        Put put3 = new Put(Bytes.toBytes("row4"));
        put3.addColumn(Bytes.toBytes("f"), Bytes.toBytes("id"), Bytes.toBytes("4"));
        put3.addColumn(Bytes.toBytes("f"), Bytes.toBytes("name"), Bytes.toBytes("bigdata"));

        List<Put> list = new ArrayList<Put>();
        list.add(put1);
        list.add(put2);
        list.add(put3);
        hTable.put(list);

        //检测put,条件成功就插入,要求RowKey是一样的
        Put put4 = new Put(Bytes.toBytes("row5"));
        put4.addColumn(Bytes.toBytes("f"), Bytes.toBytes("id"), Bytes.toBytes("5"));
        hTable.checkAndPut(Bytes.toBytes("row5"), Bytes.toBytes("f"), Bytes.toBytes("id"), null, put4);

        System.out.println("插入成功");
    }
}